package ldwu.spark.mlrec

import org.apache.log4j.{Level, Logger}
import org.apache.spark.mllib.recommendation.{ALS, MatrixFactorizationModel, Rating}
import org.apache.spark.{SparkConf, SparkContext}

object alsExamples {
  def main(args:Array[String]) {
    //屏蔽不必要的日志显示在终端上
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)

    //设置运行环境
    val sparkConf = new SparkConf().setAppName("alsExamples").setMaster("local[5]")
    val sc = new SparkContext(sparkConf)

    // Load and parse the data
    var split_str="::"
    val data = sc.textFile("/tmp/ldwu/ratings_test.dat")
    val ratings = data.map(_.split(split_str) match { case Array(user, item, rate, time) =>
      Rating(user.toInt, item.toInt, rate.toDouble)
    })

    // Build the recommendation model using ALS
    val rank = 10
    val numIterations = 10
    val model = ALS.train(ratings, rank, numIterations, 0.01)

    // Evaluate the model on rating data
    val usersProducts = ratings.map { case Rating(user, product, rate) =>
      (user, product)
    }
    val predictions =
      model.predict(usersProducts).map { case Rating(user, product, rate) =>
        ((user, product), rate)
      }
    val ratesAndPreds = ratings.map { case Rating(user, product, rate) =>
      ((user, product), rate)
    }.join(predictions)
    val MSE = ratesAndPreds.map { case ((user, product), (r1, r2)) =>
      val err = (r1 - r2)
      err * err
    }.mean()
    println("Mean Squared Error = " + MSE)

    // Save and load model
    model.save(sc, "/tmp/ldwu/myCollaborativeFilter")
    val sameModel = MatrixFactorizationModel.load(sc, "/tmp/ldwu/myCollaborativeFilter")

    // 推荐方法一
    //为每个用户进行推荐，推荐的结果可以以用户id为key，结果为value存入redis或者hbase中
    val users=data.map(_.split(split_str) match {
      case Array(user, product, rate, time) => (user)
    }).distinct().collect()

    //users: Array[String] = Array(4, 2, 3, 1)
    users.foreach(
      user => {
        //依次为用户推荐商品
        var rs = model.recommendProducts(user.toInt, numIterations)
        var value = ""
        var key = 0
        //拼接推荐结果
        rs.foreach(r => {
          key = r.user
          value = value + r.product + ":" + r.rating + ","
        })
        println(key.toString+"   " + value)
      }
    )

    //推荐方法二
    //对预测结果按预测的评分排序
    predictions.collect.sortBy(_._2)
    //对预测结果按用户进行分组，然后合并推荐结果，这部分代码待修正
    predictions.map{ case ((user, product), rate) => (user, (product,rate) )}.groupByKey.collect
    //格式化测试评分和实际评分的结果
    val formatedRatesAndPreds = ratesAndPreds.map {
      case ((user, product), (rate, pred)) => user + "," + product + "," + rate + "," + pred
    }
  }
}
